package com.xyy.saas.payment.util.mq;

import com.xyy.saas.payment.common.RocketMqGroup;
import com.xyy.saas.payment.common.RocketMqTopic;
import com.xyy.saas.payment.dao.model.ProfitSharing;
import com.xyy.saas.payment.payment.core.service.ProfitSharingService;
import com.xyy.saas.payment.util.JSONUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.Objects;

@Service
@RocketMQMessageListener(topic = RocketMqTopic.TOPIC_PROFIT_SHARING__CALLBACK, consumerGroup = RocketMqGroup.TOPIC_PROFIT_SHARING_CALLBACK_GROUP, consumeMode = ConsumeMode.CONCURRENTLY)
public class CallbackProfitSharingConsumer implements RocketMQListener<MessageExt>, RocketMQPushConsumerLifecycleListener {
    private static final Logger LOGGER = LoggerFactory.getLogger(CallbackProfitSharingConsumer.class);
    @Resource
    private ProfitSharingService profitSharingService;
    @Value("${profitSharingConsumer.switch:false}")
    private Boolean profitSharingConsumerSwitch;
    @Override
    public void onMessage(MessageExt message) {
        if (profitSharingConsumerSwitch) {
            String body = new String(message.getBody());
            LOGGER.info("CallbackProfitSharingConsumer#onMessage body:{}",  body);

            try {
                profitSharingService.profitSharing(Objects.requireNonNull(JSONUtils.parse2Object(body, ProfitSharing.class)));
            }
            catch(Exception e) {
                LOGGER.error("CallbackProfitSharingConsumer#onMessage consumer error body=" + body, e);
            }
        }
    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        consumer.setConsumeThreadMin(5);
        consumer.setConsumeThreadMax(10);
    }
}
